package src.service;

import com.alibaba.fastjson.JSON;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import src.bean.HbaseConn;
import src.bean.HbaseMap;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;


public class HbaseOperationServiceImpl implements HbaseOperationService {

    public static final String FAMILY_NAME = "f1";

    public static final byte[] CF = Bytes.toBytes(FAMILY_NAME);

    public static Connection connection = null;

    public HbaseOperationServiceImpl(HbaseConn hc) {
        connection = hc.getConnection();

    }

    public void closeConnection() {
        try {
            this.connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Override
    public List<HbaseMap> selectByTableNameAndRowkey(String tableName, String rowkey) {
        Table table = null;
        List<HbaseMap> resultList = new ArrayList<HbaseMap>();
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = table.get(get);
            Cell[] cells = result.rawCells();
            if (cells != null) {
                for (Cell cell : result.rawCells()) {
                    HbaseMap hm = new HbaseMap();
                    hm.setTableName(tableName);
                    hm.setRowkey(rowkey);
                    hm.setColumnName(new String(CellUtil.cloneQualifier(cell), "UTF-8"));
                    hm.setColumnValue(new String(CellUtil.cloneValue(cell), "UTF-8"));
                    hm.setFamily(new String(CellUtil.cloneFamily(cell), "UTF-8"));
//                    hm.setTimestamp();
                    resultList.add(hm);
                }
                return resultList;
            }
        } catch (TableNotFoundException e) {
            System.out.println("\n表" + tableName + " 不存在！");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (table != null) {
                try {
                    table.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
        return null;
    }

    @Override
    public List<HbaseMap> scanTable(String tableName) {
        List<HbaseMap> resultList = new ArrayList<HbaseMap>();
        // 装载结果集
        ResultScanner rs = null;
        try {
            TableName tn = TableName.valueOf(tableName);
            Table table_one = connection.getTable(tn);
            Scan scan = new Scan();
            rs = table_one.getScanner(scan);

            for (Result result : rs) {
                //得到单元格集合
                List<Cell> cs = result.listCells();
                for (Cell cell : cs) {
                    HbaseMap hm = new HbaseMap();
                    hm.setTableName(tableName);
                    //取行健
                    String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                    hm.setRowkey(rowKey);
                    //取到时间戳
                    long timestamp = cell.getTimestamp();
                    hm.setTimestamp(String.valueOf(timestamp));
                    //取到族列
                    String family = Bytes.toString(CellUtil.cloneFamily(cell));
                    hm.setFamily(family);
                    //取到修饰名
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    hm.setColumnName(qualifier);

                    //取到值
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    hm.setColumnValue(value);
                    resultList.add(hm);
                }
            }
            System.out.println(JSON.toJSONString(resultList));
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return resultList;
    }

    @Override
    public boolean disableTable(String tableName) {
        try {
            TableName table = TableName.valueOf(tableName);
            Admin admin = connection.getAdmin();
            if (admin.tableExists(table)) {
                admin.disableTable(table);
                return true;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean enableTable(String tableName) {
        try {
            TableName table = TableName.valueOf(tableName);
            Admin admin = connection.getAdmin();
            if (admin.isTableDisabled(table)) {
                admin.enableTable(table);
                return true;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean dropTable(String tableName) {
        try {
            TableName table = TableName.valueOf(tableName);
            Admin admin = connection.getAdmin();
            if (admin.tableExists(table)) {
                admin.disableTable(table);
                admin.deleteTable(table);
//                createTable(tableName);
                System.out.println("删除表 " + tableName + " 成功！");
                return true;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    //	通过Rowkey删除数据记录
    @Override
    public boolean deleteTableByRowkey(String tableName, String rowkey) {
        Table table = null;

        try {
            table = connection.getTable(TableName.valueOf(tableName));
            List<Delete> list = new ArrayList<Delete>();
            Delete d1 = new Delete(rowkey.getBytes());
            list.add(d1);
            table.delete(list);
            System.out.println("\n删除数据成功！ {" + tableName + "," + rowkey + "}");
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean createTable(String tableName) {
        try {
            Admin admin = connection.getAdmin();
            System.out.println("\n创建表: " + tableName);
            TableName table = TableName.valueOf(tableName);
            if (admin.tableExists(table)) {
                System.out.println(tableName + " 表已存在");
            } else {
                HTableDescriptor tableDesc = new HTableDescriptor(table);
                HColumnDescriptor columnDesc = new HColumnDescriptor(CF);
                tableDesc.addFamily(columnDesc);

                admin.createTable(tableDesc);

                System.out.println("创建表: " + tableName + " 成功");
            }
            return true;
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return false;
    }
//	public static void main(String[] args) {
//		HbaseOperationService hos = new HbaseOperationServiceImpl();
//		hos.getTableList();
//	}

    @Override
    public List<String> getTableList() {
        List<String> result = new ArrayList<String>();
        try {
            Admin admin = connection.getAdmin();
            TableName[] tables = admin.listTableNames();
            for (TableName table : tables) {
                result.add(table.getNameAsString());
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return result;
    }

    @Override
    public boolean insertData(String tableName, String rowkey, String column, String value) {
        Table table = null;
        try {
            Admin admin = connection.getAdmin();
            TableName tn = TableName.valueOf(tableName);
            table = connection.getTable(tn);
//            判断表是否存在切可用
            if (admin.tableExists(tn) && admin.isTableAvailable(tn)) {
                Put put = new Put(Bytes.toBytes(rowkey));
                if (column.contains("f1:")) {
                    System.out.println("column中包含f1,去除前三位。" + column);
                    column = column.trim().substring(3);
                }
                put.addColumn(Bytes.toBytes(FAMILY_NAME), Bytes.toBytes(column), Bytes.toBytes(value));
//                put.addColumn(Bytes.toBytes(FAMILY_NAME), Bytes.toBytes(column), Bytes.toBytes(value));
                table.put(put);
                System.out.println("表 " + tableName + " 输入数据成功! {"
                        + tableName + ", " + rowkey + ", " + column + ", " + value + "}");
            } else {
                System.out.println("表 " + tableName + " 不存在或不可用!");
            }
            return true;
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return false;
    }

    /**
     * 批量插入或修改
     *
     * @param tableName
     * @param list(rowkey:xxx,column:xxx,value:xxx)
     * @return boolean
     */
    @Override
    public boolean insertDatas(String tableName, List<Map> list) {
        Table table = null;
        try {
            Admin admin = connection.getAdmin();
            TableName tn = TableName.valueOf(tableName);
            table = connection.getTable(tn);
            List<Put> putList = new ArrayList<>();
//            判断表是否存在切可用
            if (admin.tableExists(tn) && admin.isTableAvailable(tn)) {
                for (Map map : list) {
                    Put put = new Put(Bytes.toBytes(map.get("rowkey").toString()));
                    put.addColumn(Bytes.toBytes(FAMILY_NAME), Bytes.toBytes(map.get("column").toString()),
                            Bytes.toBytes(map.get("value").toString()));
                    putList.add(put);
                }
                table.put(putList);
                System.out.println("表 " + tableName + " 输入数据成功");

            } else {
                System.out.println("表 " + tableName + " 不存在或不可用");
            }
            return true;
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean createTableSpace(String spaceName) {

        try {
            Admin admin = connection.getAdmin();
            admin.createNamespace(NamespaceDescriptor.create(spaceName).build());
            System.out.println("表空间 " + spaceName + " 创建成功！");
            return true;
        } catch (NamespaceExistException e) {
            System.out.println("表空间 " + spaceName + " 已存在！");
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean deleteTableSpace(String spaceName) {

        try {
            Admin admin = connection.getAdmin();
            TableName[] tns = admin.listTableNamesByNamespace(spaceName);
            if (tns.length > 0) {
                System.out.print("\n表空间 " + spaceName + " 下存在表数据，不可删除！\n表为：");
                for (TableName tn : tns) {
                    System.out.print(tn + "、");
                }
            } else {
                admin.deleteNamespace(spaceName);
                System.out.println("\n表空间 " + spaceName + " 删除成功！");
                return true;
            }
        } catch (NamespaceNotFoundException e) {
            System.out.println("\n表空间 " + spaceName + " 不存在！");
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }


    @Override
    public List<String> listTableSpace() {
        List<String> listNames = new ArrayList();
        String name = "";
        try {
            Admin admin = connection.getAdmin();
            NamespaceDescriptor[] spaces = admin.listNamespaceDescriptors();
            for (NamespaceDescriptor e : spaces) {
                name = e.getName();
                listNames.add(name);
//                System.out.print(e.toString() + "、");
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
        return listNames;
    }

}
